package cn.lesaas.esb.poc.route;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.JavaUuidGenerator;

import cn.lesaas.esb.poc.test.DateTimeUtils;

public class POCRoute extends RouteBuilder {
	@Override
	public void configure() throws Exception {
		// 提供TCP访问
		from("netty4:tcp://172.16.100.41:9000?allowDefaultCodec=false&encoder=#stringEncoder&decoder=#stringDecoder")
				.startupOrder(30)
				.setHeader("startTimestamp").method(DateTimeUtils.class, "toTimestamp")
				.setHeader("duration", xpath("/reqt/header/duration").resultType(int.class))
				.to("netty4:tcp://172.16.100.41:59999?allowDefaultCodec=false&encoder=#stringEncoder&decoder=#stringDecoder")				
				.choice()
					.when().xpath("/resp/header/result='FAIL'")
						.to("log:FAIL?showAll=true")
					.otherwise()
						.to("direct:insert");

		from("direct:insert").transacted().to("sql:insert into jobs(startTimestamp, duration, state, message) values (:#startTimestamp, :#duration, 0, :#${body})");

		// 轮训任务表 每隔一分钟
		from("scheduler://foo?delay=60000").to("direct:sql");

		from("direct:sql").transacted().setHeader("sequence").method(JavaUuidGenerator.class, "generateUuid")
				.to("sql:update jobs set state=1, sequence=:#sequence where state=0 and UNIX_TIMESTAMP() - UNIX_TIMESTAMP(startTimestamp)  > duration")
				.to("sql:select message from jobs where sequence=:#sequence").to("direct:Parallel");
		
		from("direct:Parallel")
			.split(body()).parallelProcessing()
				.setBody(simple("${body[message]}"))
				.log("Processing message[${property.CamelSplitIndex}]")
				.to("direct:send")
			.end();
		
		from("direct:send")
				.to("netty4:tcp://172.16.100.41:59999?allowDefaultCodec=false&encoder=#stringEncoder&decoder=#stringDecoder")
				.to("log:resp?showAll=true")
				.choice()
					.when().xpath("/resp/header/result='FAIL'")
						.to("direct:冲正")
					.otherwise()
						.to("log:SUCC?showAll=true");

		// 失败冲正交易
		from("direct:冲正").process(new Processor() {
			@Override
			public void process(Exchange exchange) throws Exception {
				System.out.println("冲正");
			}
		});
	}
}